# encoding:utf-8

# 基于 pyctp 库实现的行情 API
# 替代 AlgoPlus 库实现，保持接口兼容

import os
import csv
import threading
import time
from datetime import datetime
from typing import List, Dict, Any, Optional
#import pyctp.thostmduserapi as mdapi
from .hlog import HLog

#包含上一级目录父目录
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
import thostmduserapi as mdapi

# 创建logger实例，使用pyctp_yymmdd_hhmmss.log格式
timestamp = datetime.now().strftime("%y%m%d_%H%M")
logger = HLog(name=f"pyctp_mdapi_{timestamp}")

# 创建BarEngineCfg实例
import re
from typing import Union, TypeVar

T = TypeVar('T', str, bytes)

class AlphaPrefix:
    """通用正则：取出开头连续字母前缀，支持 str / bytes。"""
    __slots__ = ('_str_pat', '_bytes_pat')
    def __init__(self):
        self._str_pat:   Union[re.Pattern[str],  None] = None
        self._bytes_pat: Union[re.Pattern[bytes], None] = None

    def __call__(self, s: T) -> T:
        # 第一次遇到 str / bytes 时再编译，缓存下来
        if isinstance(s, str):
            if self._str_pat is None:
                self._str_pat = re.compile(r'^[A-Za-z]+')
            m = self._str_pat.match(s)
            return m.group(0) if m else ''
        elif isinstance(s, bytes):
            if self._bytes_pat is None:
                self._bytes_pat = re.compile(b'^[A-Za-z]+')
            m = self._bytes_pat.match(s)
            return m.group(0) if m else b''
        else:
            raise TypeError('need str or bytes')

# 定义BarEngineCfg类 K线切片配置参数
class BarEngineCfg:
    def __init__(self):
        #定义市场特殊时点， 除了中金所 
        self.future_market_time= {
            'start' : ["09:00:00","10:30:00","21:00:00"],   # 按照时间顺序
            'stop': ["10:15:00","11:30:00","15:00:00"],
            'call_action': ["08:59:00"]
        }
        #定义市场特殊时点，中金所
        self.future_cffex_time= {
            'start' : ["09:30:00","13:00:00"],
            'stop': ["11:30:00","15:00:00"],
            'call_action': ["09:15:00","09:25:00","09:29:00"]
        }
        # 定义CFFEX产品代码
        self.future_cffex_productid_list = {"IF","IC","IH","IM","T","TF","TS","TL"}


def to_str(data):
    """将字节数据转换为字符串"""
    if isinstance(data, bytes):
        return data.decode('utf-8', errors='ignore').rstrip('\x00')
    return str(data)


def to_bytes(data):
    """将字符串转换为字节数据"""
    if isinstance(data, str):
        return data.encode('utf-8')
    return data


class MdApiBase(mdapi.CThostFtdcMdSpi):
    """基于 pyctp 的行情 API 基础类，替代 AlgoPlus.CTP.MdApiBase"""
    
    def __init__(self, broker_id, md_server, investor_id, password, app_id, auth_code, 
                 instrument_id_list, md_queue_list=None, page_dir='', using_udp=False, multicast=False):
        super().__init__()
        
        # 基本参数
        self.broker_id = broker_id
        self.md_server = md_server
        self.investor_id = investor_id
        self.password = password
        self.app_id = app_id
        self.auth_code = auth_code
        self.instrument_id_list = instrument_id_list
        self.md_queue_list = md_queue_list or []
        self.page_dir = page_dir
        self.using_udp = using_udp
        self.multicast = multicast
        
        # 状态变量
        self.is_connected = False
        self.is_logged_in = False
        self.request_id = 0
        
        # 创建 API 实例
        self.md_api = mdapi.CThostFtdcMdApi.CreateFtdcMdApi(page_dir, using_udp, multicast)
        self.md_api.RegisterSpi(self)
        self.md_api.RegisterFront(md_server)
        
        # 初始化额外设置
        self.init_extra()
        
        # 启动 API
        self.md_api.Init()
    
    def init_extra(self):
        """子类可重写此方法进行额外初始化"""
        pass
    
    def get_next_request_id(self):
        """获取下一个请求ID"""
        self.request_id += 1
        return self.request_id
    
    def write_log(self, msg, data=None):
        """写日志"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        log_msg = f"[{timestamp}] {msg}"
        if data:
            log_msg += f" Data: {data}"
        logger.info(log_msg)
    
    def GetTradingDay(self):
        """获取交易日"""
        if hasattr(self.md_api, 'GetTradingDay'):
            return self.md_api.GetTradingDay()
        return time.strftime("%Y%m%d")
    
    def Join(self):
        """等待线程结束"""
        return self.md_api.Join()
    
    def exit(self):
        """退出API"""
        self.md_api.Release()
        
    # CTP 回调方法
    def OnFrontConnected(self):
        """前置机连接成功回调"""
        self.is_connected = True
        self.write_log("行情前置机连接成功")
        
        logger.info("行情前置机连接成功")

        # 发送登录请求
        login_req = mdapi.CThostFtdcReqUserLoginField()
        login_req.BrokerID = self.broker_id
        login_req.UserID = self.investor_id
        login_req.Password = self.password
        
        ret = self.md_api.ReqUserLogin(login_req, self.get_next_request_id())
        if ret != 0:
            self.write_log(f"发送登录请求失败，错误码: {ret}")
    
    def OnFrontDisconnected(self, nReason):
        """前置机断开连接回调"""
        self.is_connected = False
        self.is_logged_in = False
        self.write_log(f"行情前置机断开连接，原因: {nReason}")
    
    def OnRspUserLogin(self, pRspUserLogin, pRspInfo, nRequestID, bIsLast):
        """登录响应回调"""
        if pRspInfo and pRspInfo.ErrorID != 0:
            self.write_log(f"登录失败: {to_str(pRspInfo.ErrorMsg)}")
            return
        
        self.is_logged_in = True
        self.write_log("登录成功")
        
        # 订阅行情
        if self.instrument_id_list:
            # 转换合约列表为字符串列表
            instruments = []
            for instrument in self.instrument_id_list:
                if isinstance(instrument, bytes):
                    instruments.append(to_str(instrument))
                else:
                    instruments.append(str(instrument))
            
            ret = self.md_api.SubscribeMarketData(instruments, len(instruments))
            if ret != 0:
                self.write_log(f"订阅行情失败，错误码: {ret}")
            else:
                self.write_log(f"订阅行情成功，合约: {instruments}")
    
    def OnRspError(self, pRspInfo, nRequestID, bIsLast):
        """错误响应回调"""
        if pRspInfo:
            self.write_log(f"收到错误响应: {to_str(pRspInfo.ErrorMsg)}")
    
    def OnRspSubMarketData(self, pSpecificInstrument, pRspInfo, nRequestID, bIsLast):
        """订阅行情响应回调"""
        if pRspInfo and pRspInfo.ErrorID != 0:
            self.write_log(f"订阅行情失败: {to_str(pRspInfo.ErrorMsg)}")
        else:
            if pSpecificInstrument:
                self.write_log(f"订阅行情成功: {to_str(pSpecificInstrument.InstrumentID)}")
    
    def OnRtnDepthMarketData(self, pDepthMarketData):
        """深度行情通知回调 - 子类应重写此方法"""
        pass


class TickEngine(MdApiBase):
    """Tick 行情引擎"""
    
    def __init__(self, broker_id, md_server, investor_id, password, app_id, auth_code, 
                 instrument_id_list, md_queue_list=None, page_dir='', using_udp=False, multicast=False):
        super().__init__(broker_id, md_server, investor_id, password, app_id, auth_code,
                        instrument_id_list, md_queue_list, page_dir, using_udp, multicast)

    def OnRtnDepthMarketData(self, pDepthMarketData):
        """深度行情通知回调"""
        if not pDepthMarketData:
            return


            
        # 转换为字典格式，保持与 AlgoPlus 兼容
        market_data = pDepthMarketData.to_dict()
        
        # 将行情放入共享队列
        for md_queue in self.md_queue_list:
            md_queue.put(market_data)


def tick_to_bar(bar_cache, raw_frame, is_new_bar):
    # print(raw_frame['InstrumentID'], raw_frame['UpdateTime'], raw_frame['LastPrice'])
    if raw_frame['TradingDay'] != bar_cache['TradingDay']:
        bar_cache['TradingDay'] = raw_frame['TradingDay']

    tick_volume = raw_frame['Volume'] - bar_cache['DayVolume']  # Tick成交量
    bar_cache['UpdateTime'] = raw_frame['UpdateTime']  # 时间戳
    bar_cache['DayVolume'] = raw_frame['Volume']
    bar_cache['DayTurnover'] += raw_frame['Turnover']  # Day成交额
    if bar_cache['DayVolume'] > 0:
        bar_cache['DaySettlement'] = bar_cache['DayTurnover'] / bar_cache['DayVolume']  # Day均价
    bar_cache['OpenInterest'] = raw_frame['OpenInterest']  # 持仓量

    if is_new_bar:
        # 'B'为主动买，'S'为主动卖，'F'为模糊状态
        bar_cache['BVolume'] = 0
        bar_cache['SVolume'] = 0
        bar_cache['FVolume'] = 0

        bar_cache['BarVolume'] = 0  # Bar成交量
        bar_cache['BarTurnover'] = 0.0  # Bar成交额

    # 有成交
    if tick_volume > 0:
        tick_turnover = tick_volume * raw_frame['LastPrice']  # Tick成交额

        # 'B'为主动买，'S'为主动卖，'F'为模糊状态
        if raw_frame['LastPrice'] >= raw_frame['AskPrice1']:
            bar_cache['BVolume'] += tick_volume  # Bar主动买成交量
        elif raw_frame['LastPrice'] <= raw_frame['BidPrice1']:
            bar_cache['SVolume'] += tick_volume  # Bar主动卖成交量
        else:
            bar_cache['FVolume'] += tick_volume  # Bar模糊成交量

        if bar_cache['BarVolume'] == 0:
            bar_cache['LastPrice'] = raw_frame['LastPrice']  # Bar收盘价
            bar_cache['HighPrice'] = raw_frame['LastPrice']  # Bar最高价
            bar_cache['LowPrice'] = raw_frame['LastPrice']  # Bar最低价
            bar_cache['OpenPrice'] = raw_frame['LastPrice']  # Bar开盘价
            bar_cache['BarVolume'] = tick_volume  # Bar成交量
            bar_cache['BarTurnover'] = tick_turnover  # Bar成交额
            bar_cache['BarSettlement'] = raw_frame['LastPrice']  # Bar均价
        else:
            bar_cache['LastPrice'] = raw_frame['LastPrice']  # Bar收盘价
            bar_cache['HighPrice'] = max(raw_frame['LastPrice'], bar_cache['HighPrice'])  # Bar最高价
            bar_cache['LowPrice'] = min(raw_frame['LastPrice'], bar_cache['LowPrice'])  # Bar最低价
            bar_cache['BarVolume'] += tick_volume  # Bar成交量
            bar_cache['BarTurnover'] += tick_turnover  # Bar成交额
            bar_cache['BarSettlement'] = bar_cache['BarTurnover'] / bar_cache['BarVolume']  # Bar均价


class BarEngine(MdApiBase):
    """Bar 行情引擎"""
    
    def __init__(self, broker_id, md_server, investor_id, password, app_id, auth_code, 
                 instrument_id_list, md_queue_list=None, page_dir='', using_udp=False, multicast=False):
        # 注意：参数顺序与原版保持一致
        super().__init__(broker_id, md_server, investor_id, password, app_id, auth_code,
                        instrument_id_list, md_queue_list, page_dir, using_udp, multicast)
        
        # 定义CFFEX产品代码
        self.get_prefix = AlphaPrefix()
        self.cfg = BarEngineCfg()

    def init_extra(self):
        """初始化 Bar 相关数据结构"""
        # Bar字段
        bar_cache = {
            "UpdateTime": "99:99:99",
            "LastPrice": 0.0,
            "HighPrice": 0.0,
            "LowPrice": 0.0,
            "OpenPrice": 0.0,
            "BarVolume": 0,
            "BarTurnover": 0.0,
            "BarSettlement": 0.0,
            "BVolume": 0,
            "SVolume": 0,
            "FVolume": 0,
            "DayVolume": 0,
            "DayTurnover": 0.0,
            "DaySettlement": 0.0,
            "OpenInterest": 0.0,
            "LastVolume": 0,
            "TradingDay": "99999999",
        }

        self.bar_dict = {}  # Bar字典容器
        # 遍历订阅列表
        for instrument_id in self.instrument_id_list:
            # 将str转为byte
            if isinstance(instrument_id, bytes):
                instrument_id = str(instrument_id)
            # 初始化Bar字段
            self.bar_dict[instrument_id] = bar_cache.copy()

    def OnRtnDepthMarketData(self, pDepthMarketData):
        """深度行情通知回调"""
        if not pDepthMarketData:
            return
        # 转换为字典格式，保持与原版兼容
        market_data = pDepthMarketData.to_dict()
        # 转换为字节格式的合约ID
        instrument_id = market_data['InstrumentID']
        
        if instrument_id not in self.bar_dict:
            return
            
        updatetime =    market_data['UpdateTime']
        last_update_time = self.bar_dict[instrument_id]["UpdateTime"]
        
        product_id = self.get_prefix(instrument_id)
        if len(product_id) == 0:
            logger.error("Error: product_id is empty")
            return   
        t_cfg = self.cfg.future_cffex_time if product_id in self.cfg.future_cffex_productid_list else self.cfg.future_market_time
        market_start = updatetime in t_cfg['start']
        market_stop = updatetime in t_cfg['stop']
        call_action = updatetime in t_cfg.get('call_action', [])
        is_new_1minute = (updatetime[:-2] != last_update_time[:-2]) and ((not market_start) or market_stop)   # 1分钟K线条件
    
        #市场停止
        if market_stop:
            # 将Tick池化为Bar
            tick_to_bar(self.bar_dict[instrument_id], market_data, False)
            for md_queue in self.md_queue_list:
                md_queue.put(copy.deepcopy(self.bar_dict[instrument_id]))

        else:
        # # 新K线开始
            if is_new_1minute and last_update_time != "99:99:99" and not call_action:  # 实盘对于夜盘收盘的tick 产生的 K线要忽略 
                for md_queue in self.md_queue_list:
                    md_queue.put(copy.deepcopy(self.bar_dict[instrument_id]))       
            # 将Tick池化为Bar
            tick_to_bar(self.bar_dict[instrument_id], market_data, is_new_1minute)

 


class MdRecorder(MdApiBase):
    """行情记录器"""
    
    def __init__(self, broker_id, md_server, investor_id, password, app_id, auth_code, 
                 instrument_id_list, md_queue_list=None, page_dir='', using_udp=False, multicast=False):
        super().__init__(broker_id, md_server, investor_id, password, app_id, auth_code,
                        instrument_id_list, md_queue_list, page_dir, using_udp, multicast)

    def init_extra(self):
        """初始化 CSV 文件和写入器"""
        self.csv_file_dict = {}
        self.csv_writer = {}
        # 深度行情结构体字段名列表

        # 创建实例
        data = mdapi.CThostFtdcDepthMarketDataField()
        
        # 测试空实例的 to_dict
        logger.info("\n1. 测试空实例:")
        result = data.to_dict()

        header = list(result.keys())
        for instrument_id in self.instrument_id_list:
            instrument_id = to_str(instrument_id)
            # file object
            file_dir = os.path.join(self.page_dir, f'{instrument_id}-{to_str(self.GetTradingDay())}.csv')
            self.csv_file_dict[instrument_id] = open(file_dir, 'a', newline='')
            # writer object
            self.csv_writer[instrument_id] = csv.DictWriter(self.csv_file_dict[instrument_id], header)
            # 写入表头
            self.csv_writer[instrument_id].writeheader()
            self.csv_file_dict[instrument_id].flush()

    def OnRtnDepthMarketData(self, pDepthMarketData):
        """深度行情通知回调"""
        if not pDepthMarketData:
            return

 

            
        try:
            # 转换为字典格式
            market_data = pDepthMarketData.to_dict()
            
            # 写入行情
            instrument_id = market_data['InstrumentID']
            if instrument_id in self.csv_writer:
                self.csv_writer[instrument_id].writerow(market_data)
                self.csv_file_dict[instrument_id].flush()
        except Exception as err_msg:
            self.write_log(err_msg, market_data)


class FutureAccount:
    """期货账户类，用于兼容 AlgoPlus"""
    
    def __init__(self, broker_id, server_dict, investor_id, password, app_id, auth_code, 
                 instrument_id_list, md_page_dir=''):
        self.broker_id = broker_id
        self.server_dict = server_dict
        self.investor_id = investor_id
        self.password = password
        self.app_id = app_id
        self.auth_code = auth_code
        self.instrument_id_list = instrument_id_list
        self.md_page_dir = md_page_dir


def run_api(api_cls, account, md_queue_list=None):
    """运行 API"""
    if isinstance(account, FutureAccount):
        tick_engine = api_cls(
            account.broker_id,
            account.server_dict['MDServer'],
            account.investor_id,
            account.password,
            account.app_id,
            account.auth_code,
            account.instrument_id_list,
            md_queue_list,
            account.md_page_dir
        )
        tick_engine.Join()


def run_tick_engine(account, md_queue_list):
    """运行 Tick 引擎"""
    run_api(TickEngine, account, md_queue_list)


def run_bar_engine(account, md_queue_list):
    """运行 Bar 引擎"""
    run_api(BarEngine, account, md_queue_list)


def run_mdrecorder(account):
    """运行行情记录器"""
    run_api(MdRecorder, account, None)